package com.bsj.travel.mqtt.configure;

import cn.hutool.core.net.NetUtil;
import com.alibaba.fastjson.JSON;
import com.bsj.travel.constants.RedisConstant;
import com.bsj.travel.def.mqtt.message.MqttConnectOfflineMsg;
import com.bsj.travel.mqtt.enums.RedisKeys;
import com.bsj.travel.util.ExceptionUtil;
import com.bsj.travel.util.ip.GetEth0IpUtil;
import lombok.extern.slf4j.Slf4j;
import net.dreamlu.iot.mqtt.core.server.MqttServer;
import net.dreamlu.mica.core.utils.INetUtil;
import net.dreamlu.mica.redis.cache.MicaRedisCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

/**
 * @author yinhao
 * @version 1.0
 * @description 用于监听redis消息，管理mqtt集群
 * @date 2024/5/13
 */
@Slf4j
@Component
public class MqttConnectRedisListener implements MessageListener {

    @Autowired
    private MicaRedisCache micaRedisCache;

    @Autowired
    private MqttServer mqttServer;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            // 获取频道名称
            String channel = new String(message.getChannel());
            String body = new String(message.getBody());
            log.info("redis消息订阅 channel： {}, body:{}", channel, body);
            if (channel.startsWith(RedisConstant.LINE_PUB_SUB_TOPIC)) {
                MqttConnectOfflineMsg mqttConnectOfflineMsg = JSON.parseObject(body, MqttConnectOfflineMsg.class);
                if (mqttConnectOfflineMsg != null) {
                    // 删除该节点下，连接状态
                    micaRedisCache.hDel(RedisKeys.CONNECT_STATUS.getKey(GetEth0IpUtil.getHostAddress()), mqttConnectOfflineMsg.getUsername());
                    // 关闭连接
                    mqttServer.close(mqttConnectOfflineMsg.getClientId());
                }
            }
        } catch (Exception e) {
            log.error("Redis 消息订阅失败： {}", ExceptionUtil.getStackStr(e));
        }
    }
}
